package com.gsm.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * TODO
 *
 * @author gsm
 * @version 1.0
 */
public class KafkaSourceDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 为了演示方便

        // TODO 从Kafka读： 新Source架构
//        会报错 没有指定读出来数据是什么类型 跟进去源代码 发现这是一个泛型方法 一般写在builder()前面即可
//        new SimpleStringSchema() 时候会发现有问题 因为并不知道是String
//        KafkaSource<String> kafkaSource = KafkaSource.builder()
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092") // 指定kafka节点的地址和端口
                .setGroupId("gsm")  // 指定消费者组的id
                .setTopics("topic_1")   // 指定消费的 Topic
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定 反序列化器，这个是反序列化value , kafka是<key value> 但是key一般不用
                .setStartingOffsets(OffsetsInitializer.latest())  // flink消费kafka的策略
                .build();


        env
                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkasource")
//                .fromSource(kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource")
                .print();


        env.execute();
    }
}
/**
 *   kafka消费者的参数：
 *      auto.reset.offsets
 *          earliest: 如果有offset，从offset继续消费; 如果没有offset，从 最早 消费
 *          latest  : 如果有offset，从offset继续消费; 如果没有offset，从 最新 消费
 *
 *   flink的kafkasource，offset消费策略：OffsetsInitializer，默认是 earliest 可以跟下源码
 *          earliest: 一定从 最早 消费
 *          latest  : 一定从 最新 消费
 */


// 测试
// hadoop102启动zk集群 zk.sh start
// hadoop102启动kafka集群 kf.sh start
// hadoop102启动一个 kafka生产者
//      kafka-console-producer.sh --broker-list hadoop102:9092 --topic topic_1
// 输入 s1,1,1
//[root@hadoop102 ~]# kafka-console-producer.sh --broker-list hadoop102:9092 --topic topic_1
//>s1,1,1
//[2024-10-13 17:16:19,139] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 4 : {topic_1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
//以上警告是正常的 因为没有提前创建这个topic 接着启动当前flink程序后, 再输入
//>s2,2,2

//观察idea控制台输出
//"C:\Program Files\Java\jdk1.8.0_151\bin\java.exe" -javaagent:D:\ProgramFiles\JetBrains\IntelliJIDEA2023.1.4\lib\idea_rt.jar=47826:D:\ProgramFiles\JetBrains\IntelliJIDEA2023.1.4\bin -Dfile.encoding=UTF-8 -classpath "C:\Program Files\Java\jdk1.8.0_151\jre\lib\charsets.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\deploy.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\access-bridge-64.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\cldrdata.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\dnsns.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\jaccess.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\jfxrt.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\localedata.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\nashorn.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\sunec.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\sunjce_provider.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\sunmscapi.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\sunpkcs11.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\ext\zipfs.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\javaws.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\jce.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\jfr.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\jfxswt.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\jsse.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\management-agent.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\plugin.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\resources.jar;C:\Program Files\Java\jdk1.8.0_151\jre\lib\rt.jar;D:\IdeaProjectsBigData\FlinkTutorial-1.17\target\classes;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-streaming-java\1.17.0\flink-streaming-java-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-core\1.17.0\flink-core-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-annotations\1.17.0\flink-annotations-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-metrics-core\1.17.0\flink-metrics-core-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-shaded-asm-9\9.3-16.1\flink-shaded-asm-9-9.3-16.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\commons\commons-lang3\3.12.0\commons-lang3-3.12.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\commons\commons-text\1.10.0\commons-text-1.10.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\com\esotericsoftware\kryo\kryo\2.24.0\kryo-2.24.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\com\esotericsoftware\minlog\minlog\1.2\minlog-1.2.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\objenesis\objenesis\2.1\objenesis-2.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\commons-collections\commons-collections\3.2.2\commons-collections-3.2.2.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\commons\commons-compress\1.21\commons-compress-1.21.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-file-sink-common\1.17.0\flink-file-sink-common-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-runtime\1.17.0\flink-runtime-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-rpc-core\1.17.0\flink-rpc-core-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-rpc-akka-loader\1.17.0\flink-rpc-akka-loader-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-queryable-state-client-java\1.17.0\flink-queryable-state-client-java-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-hadoop-fs\1.17.0\flink-hadoop-fs-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\commons-io\commons-io\2.11.0\commons-io-2.11.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-shaded-zookeeper-3\3.7.1-16.1\flink-shaded-zookeeper-3-3.7.1-16.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\javassist\javassist\3.24.0-GA\javassist-3.24.0-GA.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\xerial\snappy\snappy-java\1.1.8.3\snappy-java-1.1.8.3.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\lz4\lz4-java\1.8.0\lz4-java-1.8.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-java\1.17.0\flink-java-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\com\twitter\chill-java\0.7.6\chill-java-0.7.6.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-shaded-guava\30.1.1-jre-16.1\flink-shaded-guava-30.1.1-jre-16.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\commons\commons-math3\3.6.1\commons-math3-3.6.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\slf4j\slf4j-api\1.7.36\slf4j-api-1.7.36.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-clients\1.17.0\flink-clients-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-optimizer\1.17.0\flink-optimizer-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\commons-cli\commons-cli\1.5.0\commons-cli-1.5.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-runtime-web\1.17.0\flink-runtime-web-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-shaded-netty\4.1.82.Final-16.1\flink-shaded-netty-4.1.82.Final-16.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-shaded-jackson\2.13.4-16.1\flink-shaded-jackson-2.13.4-16.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-connector-files\1.17.0\flink-connector-files-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-shaded-force-shading\16.1\flink-shaded-force-shading-16.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-connector-kafka\1.17.0\flink-connector-kafka-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-connector-base\1.17.0\flink-connector-base-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\kafka\kafka-clients\3.2.3\kafka-clients-3.2.3.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\com\github\luben\zstd-jni\1.5.2-1\zstd-jni-1.5.2-1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-connector-datagen\1.17.0\flink-connector-datagen-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\mysql\mysql-connector-java\8.0.27\mysql-connector-java-8.0.27.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\com\google\protobuf\protobuf-java\3.11.4\protobuf-java-3.11.4.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-connector-jdbc\1.17-SNAPSHOT\flink-connector-jdbc-1.17-20230109.003314-120.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-statebackend-changelog\1.17.0\flink-statebackend-changelog-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-statebackend-common\1.17.0\flink-statebackend-common-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\com\google\code\findbugs\jsr305\1.3.9\jsr305-1.3.9.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-table-api-java-bridge\1.17.0\flink-table-api-java-bridge-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-table-api-java\1.17.0\flink-table-api-java-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-table-api-bridge-base\1.17.0\flink-table-api-bridge-base-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-table-planner-loader\1.17.0\flink-table-planner-loader-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-table-runtime\1.17.0\flink-table-runtime-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-table-common\1.17.0\flink-table-common-1.17.0.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\com\ibm\icu\icu4j\67.1\icu4j-67.1.jar;D:\ProgramFiles\apache-maven-3.5.4\repository\org\apache\flink\flink-cep\1.17.0\flink-cep-1.17.0.jar" com.gsm.source.KafkaSourceDemo
//SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
//SLF4J: Defaulting to no-operation (NOP) logger implementation
//SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
//s2,2,2

//可以继续测试
//>s3,3,3
//>s4,4,4